package com.xjd.transport.impl;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.xjd.io.in.EventInputStream;
import com.xjd.transport.Channel;
import com.xjd.transport.Future;
import com.xjd.transport.ReceiveHandler;
import com.xjd.transport.TransportException;
import com.xjd.transport.TransportRuntimeException;
import com.xjd.transport.TransportStates;
import com.xjd.transport.work.ReceiveThread;
import com.xjd.transport.work.SendWork;
import com.xjd.transport.work.WorkExecutor;

public class ChannelImpl implements Channel, TransportStates {
	ReentrantReadWriteLock stateReadWriteLock = new ReentrantReadWriteLock();
	int state = STATE_INITIAL;

	SocketChannel socketChannel;
	ExecutorService jobService;

	Object sendLock = new Object();
	WorkExecutor sendWorkExecutor;
	byte[] sendBuf = new byte[1024 * 2]; //2k
	
	Object receiveLock = new Object();
	ReceiveStream receiveStream = null;

	TransportAdapter adapter;

	// ==static open
	public static ChannelImpl open(String host, int port) throws IOException {
		return open(new InetSocketAddress(host, port));
	}

	public static ChannelImpl open(InetSocketAddress dest) throws IOException {
		return open(dest, Executors.newCachedThreadPool());
	}

	public static ChannelImpl open(InetSocketAddress dest, ExecutorService jobService) throws IOException {
		SocketChannel socketChannel = SocketChannel.open(dest);
		return new ChannelImpl(socketChannel, jobService);
	}

	// ==constructor
	public ChannelImpl(SocketChannel socketChannel, ExecutorService jobService) {
		this(socketChannel, jobService, null);
	}

	public ChannelImpl(SocketChannel socketChannel, ExecutorService jobService, TransportAdapter adapter) {
		this.socketChannel = socketChannel;
		this.jobService = jobService;
		this.adapter = adapter;
		state = STATE_STARTED;
	}

	// ==overide
	@Override
	public InetSocketAddress getLocalAddress() {
		return (InetSocketAddress) socketChannel.socket().getLocalSocketAddress();
	}

	@Override
	public InetSocketAddress getRemoteAddress() {
		return (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
	}

	protected void checkStopped() {
		stateReadWriteLock.readLock().lock();
		try {
			if (state >= STATE_STOPED) {
				throw new TransportRuntimeException("Channel has been stopped.");
			}
		} finally {
			stateReadWriteLock.readLock().unlock();
		}
	}

	@Override
	public void send(InputStream src) throws TransportException {
		checkStopped();
		if (sendWorkExecutor == null) {
			synchronized (sendLock) {
				if (sendWorkExecutor == null) {
					initSendWorkExecutor();
				}
			}
		}
		FutureImpl future = new FutureImpl(this);
		try {
			SendWork work = new SendWork(src, socketChannel.socket().getOutputStream(), this.sendBuf, future);
			sendWorkExecutor.execute(work, false);
		} catch (IOException e) {
			throw new TransportException(e);
		}
		if (future.isFailed()) {
			Throwable cause = future.getCause();
			if (cause instanceof RuntimeException) {
				throw (RuntimeException) cause;
			} else if (cause instanceof TransportException) {
				throw (TransportException) cause;
			} else {
				throw new TransportException(future.getCause());
			}
		}
	}

	@Override
	public Future asyncSend(InputStream src) {
		checkStopped();
		if (sendWorkExecutor == null) {
			synchronized (sendLock) {
				if (sendWorkExecutor == null) {
					initSendWorkExecutor();
				}
			}
		}
		FutureImpl future = new FutureImpl(this);
		try {
			SendWork work = new SendWork(src, socketChannel.socket().getOutputStream(), this.sendBuf, future);
			sendWorkExecutor.execute(work, true);
		} catch (IOException e) {
			future.setState(STATE_STOPED);
			future.setFinishState(FINISH_STATE_FAILED);
			future.setCause(e);
		}
		return future;
	}

	protected void initSendWorkExecutor() {
		sendWorkExecutor = new WorkExecutor(this.jobService);
	}
	
	protected void initReceiveStream() throws IOException {
		receiveStream = new ReceiveStream(this.socketChannel.socket().getInputStream());
	}

	@Override
	public InputStream receive() throws TransportException {
		checkStopped();
		synchronized (receiveLock) {
			if (receiveStream != null && !receiveStream.getFuture().isDone()) {
				throw  new TransportException("'receiving' in progress by another thread.");
			}
			if (receiveStream == null) {
				try {
					initReceiveStream();
				} catch (IOException e) {
					throw new TransportException(e);
				}
			}
			FutureImpl future = new FutureImpl(this);
			receiveStream.setFuture(future);
			return receiveStream;
		}
	}

	@Override
	public Future asyncReceive(ReceiveHandler handler) {
		checkStopped();
		synchronized (receiveLock) {
			FutureImpl future = new FutureImpl(this);
			Throwable e = null;
			if (receiveStream != null && !receiveStream.getFuture().isDone()) {
				e = new TransportException("'receiving' in progress by another thread.");
			} else {
				//TODO 起动线程去执行接收
				if (receiveStream == null) {
					try {
						initReceiveStream();
					} catch (IOException ex) {
						e = ex;
					}
				}
				if (e == null) {
					receiveStream.setFuture(future);
					ReceiveThread rt = new ReceiveThread(receiveStream, handler);
					if (jobService == null) {
						rt.start();
					} else {
						jobService.execute(rt);
					}
				}
			}
			if (e != null) {
				synchronized (future.getStateLock()) {
					future.setState(STATE_STOPED);
					future.setFinishState(FINISH_STATE_FAILED);
					future.setCause(e);
					future.getStateLock().notifyAll();
				}
				future.fireFutureListener();
			}
			return future;
		}
	}

	@Override
	public void close() {
		stateReadWriteLock.writeLock().lock();
		try {
			if (state == STATE_STOPED) {
				return;
			}
			state = STATE_STOPED;
		} finally {
			stateReadWriteLock.writeLock().unlock();
		}
		try {
			socketChannel.close();
		} catch (IOException e) {
			// DoNothing
		}
		adapter.channelClosed(this);
	}

	public static interface TransportAdapter {
		void channelClosed(ChannelImpl channel);
	}
	
	public static class ReceiveStream extends EventInputStream {
		FutureImpl future;

		public ReceiveStream(InputStream in) {
			super(in);
			addListener(new EventInputStreamListenerAdapter() {
				public void onException(Thread useThread, IOException e) {
					FutureImpl f = getFuture();
					if (f != null) {
						synchronized (f.getStateLock()) {
							if (f.isDone()) {
								return;
							}
							f.setState(STATE_STOPED);
							f.setFinishState(FINISH_STATE_FAILED);
							f.setCause(e);
							f.getStateLock().notifyAll();
						}
						f.fireFutureListener();
					}
				}

				public boolean beforeClose(Thread useThread) throws IOException {
					boolean rt = false;
					FutureImpl f = getFuture();
					if (f != null) {
						synchronized (f.getStateLock()) {
							if (f.isDone()) {
								return rt;
							}
							f.setState(STATE_STOPED);
							f.setFinishState(FINISH_STATE_SUCCESS);
							f.getStateLock().notifyAll();
						}
						f.fireFutureListener();
					}
					return rt;
				}

				@Override
				public int afterRead(Thread useThread, int readInt) throws IOException {
					if (readInt != -1) {
						synchronized (future.getDoneBytesLock()) {
							future.setDoneBytes(future.getDoneBytes() + 1);
						}
					}
					return super.afterRead(useThread, readInt);
				}

				@Override
				public void afterReadBytes(Thread useThread, byte[] b, int off, int readLen) throws IOException {
					synchronized (future.getDoneBytesLock()) {
						future.setDoneBytes(future.getDoneBytes() + readLen);
					}
				}

				@Override
				public void afterSkip(Thread useThread, long skipped) throws IOException {
					synchronized (future.getDoneBytesLock()) {
						future.setDoneBytes(future.getDoneBytes() + skipped);
					}
				}
				
			});
		}

		public FutureImpl getFuture() {
			return future;
		}

		public void setFuture(FutureImpl future) {
			this.future = future;
		}
		
	}

}
